//===----------------------------------------------------------------------===//
//
//                         BusTub
//
// update_executor.cpp
//
// Identification: src/execution/update_executor.cpp
//
// Copyright (c) 2015-2021, Carnegie Mellon University Database Group
//
//===----------------------------------------------------------------------===//
#include <memory>
#include <vector>

#include "common/config.h"
#include "common/macros.h"
#include "common/rid.h"
#include "execution/executors/update_executor.h"
#include "storage/table/tuple.h"
#include "type/value.h"

namespace bustub {

UpdateExecutor::UpdateExecutor(ExecutorContext *exec_ctx, const UpdatePlanNode *plan,
                               std::unique_ptr<AbstractExecutor> &&child_executor)
    : AbstractExecutor(exec_ctx),
      plan_(plan),
      table_info_(exec_ctx->GetCatalog()->GetTable(plan->table_oid_)),
      child_executor_(std::move(child_executor)) {
  // As of Fall 2022, you DON'T need to implement update executor to have perfect score in project 3 / project 4.
}

void UpdateExecutor::Init() {
  sum_ = 0;
  child_executor_->Init();
}

auto UpdateExecutor::Next([[maybe_unused]] Tuple *tuple, RID *rid) -> bool {
  if (ending_) {
    return false;
  }
  std::vector<Value> output_values;
  output_values.reserve(plan_->OutputSchema().GetColumnCount());

  std::vector<Value> update_values;
  update_values.reserve(child_executor_->GetOutputSchema().GetColumnCount());
  Tuple old_tuple{};
  RID old_rid{};

  auto all_index_info = GetExecutorContext()->GetCatalog()->GetTableIndexes(table_info_->name_);

  auto status = child_executor_->Next(&old_tuple, &old_rid);
  while (status) {
    BUSTUB_ENSURE(old_rid.GetPageId() != INVALID_PAGE_ID, "invalid rid");
    // create new tuple
    for (auto &expr : plan_->target_expressions_) {
      update_values.push_back(expr->Evaluate(&old_tuple, child_executor_->GetOutputSchema()));
    }
    Tuple updated_tuple{update_values, &child_executor_->GetOutputSchema()};
    update_values.clear();
    // delete old tuple
    TupleMeta del_meta{table_info_->table_->GetTupleMeta(old_rid)};
    del_meta.is_deleted_ = true;
    table_info_->table_->UpdateTupleMeta(del_meta, old_rid);
    // delete old indexes
    for (auto index_info : all_index_info) {
      index_info->index_->DeleteEntry(
          old_tuple.KeyFromTuple(table_info_->schema_, index_info->key_schema_, index_info->index_->GetKeyAttrs()),
          old_rid, GetExecutorContext()->GetTransaction());
    }
    // insert new tuple
    auto updated_rid = table_info_->table_->InsertTuple({INVALID_TXN_ID, INVALID_TXN_ID, false}, updated_tuple);
    if (!updated_rid.has_value() || updated_rid->GetPageId() == INVALID_PAGE_ID) {
      output_values.push_back(bustub::ValueFactory::GetIntegerValue(sum_));
      Tuple output_tuple{output_values, &GetOutputSchema()};
      return false;
    }
    // insert new index
    for (auto index_info : all_index_info) {
      index_info->index_->InsertEntry(
          updated_tuple.KeyFromTuple(table_info_->schema_, index_info->key_schema_, index_info->index_->GetKeyAttrs()),
          updated_rid.value(), GetExecutorContext()->GetTransaction());
    }

    ++sum_;
    status = child_executor_->Next(&old_tuple, &old_rid);
  }

  output_values.push_back(bustub::ValueFactory::GetIntegerValue(sum_));
  Tuple output_tuple{output_values, &GetOutputSchema()};
  *tuple = output_tuple;
  ending_ = true;

  return true;
}

}  // namespace bustub
